package com.atguigu.realtime.util;

import com.atguigu.realtime.annotation.NoSink;
import com.atguigu.realtime.bean.VisitorStats;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/4/24 15:44
 */
public class MySinkUtil {
    
    public static <T> SinkFunction<T> getClickHouseSink(String db,
                                                        String table,
                                                        Class<T> tClass) {
        
        String url = "jdbc:clickhouse://hadoop162:8123/" + db;
        String driver = "ru.yandex.clickhouse.ClickHouseDriver";
        StringBuilder sql = new StringBuilder();
        
        // insert into user (id, ....) values (?, ...)
        sql.append("insert into ").append(table).append("(");
        // 找到tClass他的所有的属性
        Field[] fields = tClass.getDeclaredFields();
        for (Field field : fields) {
            NoSink noSink = field.getAnnotation(NoSink.class);
            if (noSink == null) {
                sql.append(field.getName()).append(",");
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(") values (");
        for (Field field : fields) {
            NoSink noSink = field.getAnnotation(NoSink.class);
            if (noSink == null) {
                sql.append("?,");
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(")");
        
        return getJdbcSink(url, driver, sql.toString());
    }
    
    public static void main(String[] args) {
        getClickHouseSink(null, null, VisitorStats.class);
    }
    
    public static <T> SinkFunction<T> getJdbcSink(String url,
                                                  String driver,
                                                  String sql) {
        return JdbcSink.sink(
            sql,
            new JdbcStatementBuilder<T>() {
                @Override
                public void accept(PreparedStatement ps,
                                   T t) throws SQLException {
                    System.out.println(t);
                    // TODO 给ps中的占位符进行赋值
                    Field[] fields = t.getClass().getDeclaredFields();
                    for (int i = 0, position = 1; i < fields.length; i++) { // 30个属性  跳过第27个  最多29个问号
                        Field field = fields[i];
                        
                        NoSink noSink = field.getAnnotation(NoSink.class);
                        if (noSink == null) {
                            field.setAccessible(true);
                            // 取出 t这个对象的 属性的值  t.stt  反射: field.get(t)
                            try {
                                Object obj = field.get(t);
                                ps.setObject(position++, obj);  // 28 29
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                        
                    }
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchIntervalMs(1000)
                //                .withBatchSize(1024)
                //                .withMaxRetries(5)
                .build(),
            new JdbcConnectionOptions
                .JdbcConnectionOptionsBuilder()
                .withDriverName(driver)
                .withUrl(url)
                .build()
        );
    }
    
}
